package com.rizzo.shell.intern;

import com.google.common.base.Predicate;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.rizzo.shell.plugins.AmqpConfig;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.shell.core.CommandMarker;
import org.springframework.shell.core.annotation.CliCommand;
import org.springframework.shell.core.annotation.CliOption;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;

@Component
public class AmqpCommands implements CommandMarker {

	private static Logger logger = Logger.getLogger("com.rizzo.shell.intern.AmqpCommands");

	public enum TransferArgument {
		HOST, PORT, USERNAME, PASSWORD, VHOST, TTL
	}

	private Set<AmqpLogin> amqpLogins = ImmutableSet.of(
			new AmqpLogin(1, "amqp-on-1-mgt.mmis.be", "LoggeR", "5672", "logger", Environment.ON, "logging"),
			new AmqpLogin(2, "amqp-on-2-mgt.mmis.be", "LoggeR", "5672", "logger", Environment.ON, "logging"));

	@CliCommand(value = "amqp:produce", help = "Produces an amount of messages on a RabbitMQ host")
	public String execute(
			@CliOption(key = {"node"}, mandatory = true, help = "The node to produce messages to") final int node,
			@CliOption(key = {"environment"}, mandatory = true, help = "The environment to produce messages to") final Environment environment,
			@CliOption(key = {"ttl"}, mandatory = true, help = "The ttl to produce messages to") final String ttl,
			@CliOption(key = {"amount"}, mandatory = true, help = "The amount of messages") final String amount) throws Exception {
		logger.info("Producing messages amount: " + amount + " ...");
		AmqpLogin amqpLogin = getAmqpLogin(node, environment);
		try {
			Map<AmqpCommands.TransferArgument, String> args = ImmutableMap.<AmqpCommands.TransferArgument, String>builder()
					.put(TransferArgument.HOST, amqpLogin.getHost())
					.put(TransferArgument.PORT, amqpLogin.getPort())
					.put(TransferArgument.USERNAME, amqpLogin.getUsername())
					.put(TransferArgument.PASSWORD, amqpLogin.getPassword())
					.put(TransferArgument.VHOST, amqpLogin.getVhost())
					.put(TransferArgument.TTL, ttl).build();

			BeanDefinition beanDefinition = BeanDefinitionBuilder.rootBeanDefinition(HashMap.class).addConstructorArgValue(args).getBeanDefinition();

			AnnotationConfigApplicationContext componentContext = new AnnotationConfigApplicationContext();
			componentContext.registerBeanDefinition("args", beanDefinition);
			componentContext.register(AmqpConfig.class);
			componentContext.refresh();

			RabbitTemplate template = (RabbitTemplate) componentContext.getBean("rabbitTemplate");
			template.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
				@Override
				public void confirm(CorrelationData correlationData, boolean ack) {
					logger.log(Level.INFO, "Publisher Callback called for correlationData: " + correlationData.getId() + " with ack: " + ack);
				}
			});

			for (int t = 0; t < Integer.parseInt(amount); t++) {
				final MessageProperties messageProperties = new MessageProperties();
				messageProperties.setCorrelationId(String.valueOf(t).getBytes());
				messageProperties.setDeliveryTag(t);
				messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
				template.send("logging", "my-queue-key", new Message(String.valueOf(t).getBytes(), messageProperties),
						new CorrelationData(String.valueOf(t)));
			}
		} catch (Exception e) {
			logger.log(Level.SEVERE, "Fail!", e);
		}
		return "Done producing messages";
	}


	public AmqpLogin getAmqpLogin(final int node, final Environment environment) {
		Predicate<AmqpLogin> matches = new Predicate<AmqpLogin>() {
			@Override
			public boolean apply(AmqpLogin amqpLogin) {
				boolean nodeEquals = amqpLogin.getNode() == node;
				boolean environmentEquals = amqpLogin.getEnvironment() == environment;
				return nodeEquals && environmentEquals;
			}
		};
		Iterable<AmqpLogin> filteredLogins = Iterables.filter(amqpLogins, matches);
		return Iterables.getFirst(filteredLogins, null);
	}

	private enum Environment {
		ON, OE, PR
	}

	private class AmqpLogin {

		private int node;
		private Environment environment;
		private String host;
		private String port;
		private String username;
		private String password;
		private String vhost;

		private AmqpLogin(int node, String host, String password, String port, String username, Environment environment, String vhost) {
			this.environment = environment;
			this.node = node;
			this.host = host;
			this.password = password;
			this.port = port;
			this.username = username;
			this.vhost = vhost;
		}

		public int getNode() {
			return node;
		}

		public void setNode(int node) {
			this.node = node;
		}

		public Environment getEnvironment() {
			return environment;
		}

		public void setEnvironment(Environment environment) {
			this.environment = environment;
		}

		public String getHost() {
			return host;
		}

		public void setHost(String host) {
			this.host = host;
		}

		public String getPassword() {
			return password;
		}

		public void setPassword(String password) {
			this.password = password;
		}

		public String getPort() {
			return port;
		}

		public void setPort(String port) {
			this.port = port;
		}

		public String getUsername() {
			return username;
		}

		public void setUsername(String username) {
			this.username = username;
		}

		public String getVhost() {
			return vhost;
		}

		public void setVhost(String vhost) {
			this.vhost = vhost;
		}
	}

}
